package com.atguigu.Flink.datastream.transform;

import com.atguigu.Flink.POJO.Event;
import com.atguigu.Flink.function.ClickSource;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

public class Flink08_SplitStream {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Event> ds = env.addSource(new ClickSource());
        //测试侧输出流
        OutputTag<Event> ottoAndxuanDog = new OutputTag<Event>("ottoAndxuanDog", Types.POJO(Event.class));
        SingleOutputStreamOperator<Event> process = ds.process(new ProcessFunction<Event, Event>() {
            @Override
            public void processElement(Event event, ProcessFunction<Event, Event>.Context context, Collector<Event> collector) throws Exception {
                if (event.getUser().equals("otto") || event.getUser().equals("xuanDog")) {
                    //发送到侧流
                    context.output(ottoAndxuanDog, event);
                } else {
                    //发送到当前流
                    collector.collect(event);
                }
            }
        });
        process.print("main");
        process.getSideOutput(ottoAndxuanDog).print("ottoAndxuanDog");

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }
}
